Roots 演练
交互式练习:对照下面这个完整的 MCP 聊天项目,逐步查看根目录(roots)如何在客户端定义并通过回调提供给服务器,以及服务器如何读取根目录列表并据此授权文件访问。点击"下一步"会自动打开对应文件并高亮相关代码。
教程步骤
让我们通过一个示例项目来更好地了解如何实现此功能。
理想情况下,用户将指定 MCP 服务器可以访问哪些文件/文件夹。
该程序设置为接受一个 CLI 参数列表,这些参数被解释为用户希望允许访问的路径。
该路径列表在第 42 行提供给 MCPClient。
根据 MCP 规范,所有根目录都应具有以 file:// 开头的 URI。
此函数获取用户提供的路径列表,并将它们转换为 Root 对象。
客户端不会立即将根目录列表提供给服务器。相反,服务器可以在未来某个时间点向客户端发出请求。我们创建一个回调,当服务器请求根目录时将执行该回调。该回调需要在 ListRootsResult 对象中返回根目录列表。
此回调在第 58 行传入 ClientSession。
接下来是服务器端。服务器将在两种场景中使用根目录:
- 每当工具尝试访问文件或文件夹时
- 当 LLM(如 Claude)需要将文件或文件夹解析为完整路径时。想象一下当用户说"读取 todos.txt 文件"时——Claude 需要弄清楚该文本文件在哪里,它可能通过查看根目录列表来实现这一点
为了处理第二种情况,我们可以定义一个列出根目录的工具,或者直接将它们注入到提示中。
通过调用 ctx.session.list_roots() 来访问根目录。
这会向客户端发送一条消息,从而使其运行根目录列表回调。
请记住:MCP SDK 不会尝试限制您的工具可以读取哪些文件或文件夹!您必须自己实现该检查。
考虑实现一个类似 is_path_allowed 的函数,它将通过将路径与根目录列表进行比较来决定该路径是否可访问。
一旦您构建好了一个授权函数——比如 is_path_allowed——请在您的工具中使用它,以确保请求的路径是可访问的。
import asyncioimport sysimport osfrom dotenv import load_dotenvfrom contextlib import AsyncExitStackfrom mcp_client import MCPClientfrom core.claude import Claudefrom core.cli_chat import CliChatfrom core.cli import CliAppload_dotenv()# Anthropic Configclaude_model = os.getenv("CLAUDE_MODEL", "claude-sonnet-4-5")anthropic_api_key = os.getenv("ANTHROPIC_API_KEY", "")assert claude_model, "Error: CLAUDE_MODEL cannot be empty. Update .env"assert anthropic_api_key, ("Error: ANTHROPIC_API_KEY cannot be empty. Update .env")async def main():claude_service = Claude(model=claude_model)# Get root directories from command line argumentsroot_paths = sys.argv[1:]if not root_paths:print("Usage: uv run main.py <root1> [root2] ...")print("Example: uv run main.py /path/to/videos /another/path")sys.exit(1)clients = {}async with AsyncExitStack() as stack:# Create the MCP client with the provided root directoriesdoc_client = await stack.enter_async_context(MCPClient(command="uv", args=["run", "mcp_server.py"], roots=root_paths))clients["doc_client"] = doc_clientchat = CliChat(doc_client=doc_client,clients=clients,claude_service=claude_service,)cli = CliApp(chat)await cli.initialize()await cli.run()if __name__ == "__main__":if sys.platform == "win32":asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())asyncio.run(main())